03 蓝图拆分大型Agent项目

上一篇将所有接口都集中在一个app.py中：健康检查、对话、会话管理、错误处理等。功能较少时这种方式足够简单，但随着接口数量增加，单文件会逐渐变长，维护成本也会提高。

一种直接的拆分思路是将不同功能的路由放到不同文件中。但如果继续使用@app.route()装饰器，每个路由文件都需要获取app对象，容易引入循环引用问题。

Blueprint（蓝图）是 Flask 用于解决模块化路由组织的机制：先在独立模块中定义路由，再统一注册到应用上。这样路由文件不需要直接依赖全局app对象。

通过 Blueprint，可以将 Agent API 按功能拆分为对话、会话、健康检查等模块，再在应用入口统一组装。

一、创建Blueprint

一个 Blueprint 可以理解为一个独立的路由模块，创建方式与 Flask 应用类似：

python

# routes/chat.py
from flask import Blueprint, request

# 创建Blueprint实例
# 第一个参数是蓝图名称，第二个参数通常是__name__
chat_bp = Blueprint("chat", __name__)


@chat_bp.route("/chat", methods=["POST"])
def chat():
    """对话接口"""
    data = request.get_json()
    message = data.get("message", "")
    return {"reply": f"收到: {message}"}


@chat_bp.route("/chat/models")
def list_models():
    """列出可用模型"""
    return {"models": ["deepseek-v4-flash", "deepseek-v3"]}

这里使用的是@chat_bp.route()，而不是@app.route()。Blueprint 会记录待注册的路由，但在注册到应用之前不会真正生效。

1.1 Blueprint构造参数

python

Blueprint(name, import_name, **options)

参数	说明	示例
`name`	蓝图名称，用于端点前缀	`"chat"`
`import_name`	通常传`__name__`，用于定位资源文件	`__name__`
`url_prefix`	URL前缀，所有路由都会加上这个前缀	`"/api/v1"`
`template_folder`	模板文件夹	`"templates"`
`static_folder`	静态文件夹	`"static"`

二、注册Blueprint

定义 Blueprint 之后，需要在应用中注册：

python

# app.py
from flask import Flask
from routes.chat import chat_bp

app = Flask(__name__)

# 注册蓝图
app.register_blueprint(chat_bp)

if __name__ == "__main__":
    app.run(debug=True)

注册完成后，chat_bp里定义的路由才会真正生效。

2.1 URL前缀

注册时可以指定 URL 前缀，使该蓝图下的所有路由都带上统一前缀：

python

# 所有路由前面自动加上 /api/v1
app.register_blueprint(chat_bp, url_prefix="/api/v1")

原来定义的/chat会变成/api/v1/chat，/chat/models会变成/api/v1/chat/models。

前缀也可以在创建Blueprint时就指定：

python

chat_bp = Blueprint("chat", __name__, url_prefix="/api/v1")

两种方式效果一致，可以根据项目习惯选择。

三、拆分Agent项目

下面是一个 Agent 项目中常见的拆分方式：

my-agent-api/
├── app.py                  # 应用入口，组装所有蓝图
├── routes/
│   ├── __init__.py
│   ├── chat.py             # 对话相关接口
│   ├── session.py          # 会话管理接口
│   └── health.py           # 健康检查接口
└── errors.py               # 错误处理

3.1 chat.py — 对话接口

python

# routes/chat.py
from flask import Blueprint, request, abort

chat_bp = Blueprint("chat", __name__)


@chat_bp.route("/chat", methods=["POST"])
def chat():
    """发送消息，获取Agent回复"""
    data = request.get_json()
    if not data or "message" not in data:
        abort(400, description="缺少message字段")

    message = data["message"]
    session_id = data.get("session_id", "default")

    # 后面会替换成真正的Agent调用
    return {
        "reply": f"收到: {message}",
        "session_id": session_id,
    }


@chat_bp.route("/chat/models")
def list_models():
    """列出可用模型"""
    return {
        "models": [
            {"id": "deepseek-v4-flash", "name": "DeepSeek V4 Flash"},
            {"id": "deepseek-v3", "name": "DeepSeek V3"},
        ]
    }

3.2 session.py — 会话管理

python

# routes/session.py
from flask import Blueprint, jsonify

session_bp = Blueprint("session", __name__)


@session_bp.route("/session/<session_id>")
def get_session(session_id):
    """获取会话信息"""
    return {
        "session_id": session_id,
        "messages": [],
        "created_at": "2026-01-01T00:00:00Z",
    }


@session_bp.route("/session/<session_id>", methods=["DELETE"])
def delete_session(session_id):
    """删除会话"""
    return {"deleted": session_id}


@session_bp.route("/sessions")
def list_sessions():
    """列出所有会话"""
    return {"sessions": []}

3.3 health.py — 健康检查

python

# routes/health.py
from flask import Blueprint

health_bp = Blueprint("health", __name__)


@health_bp.route("/health")
def health():
    """健康检查"""
    return {"status": "ok"}


@health_bp.route("/version")
def version():
    """版本信息"""
    return {"version": "1.0.0"}

3.4 errors.py — 错误处理

python

# errors.py
from flask import Flask, jsonify


def register_error_handlers(app: Flask):
    """注册全局错误处理器"""

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({"error": str(error.description)}), 400

    @app.errorhandler(404)
    def not_found(error):
        return jsonify({"error": "接口不存在"}), 404

    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({"error": "服务器内部错误"}), 500

3.5 app.py — 组装一切

python

# app.py
from flask import Flask
from routes.chat import chat_bp
from routes.session import session_bp
from routes.health import health_bp
from errors import register_error_handlers


def create_app():
    """创建Flask应用"""
    app = Flask(__name__)

    # 注册蓝图
    app.register_blueprint(chat_bp, url_prefix="/api")
    app.register_blueprint(session_bp, url_prefix="/api")
    app.register_blueprint(health_bp)

    # 注册错误处理
    register_error_handlers(app)

    return app


if __name__ == "__main__":
    app = create_app()
    app.run(debug=True)

此时，app.py只负责应用创建、蓝图注册和错误处理注册。各接口模块只处理自身功能，模块边界更清晰。

最终 API 列表如下：

方法	路径	来源
GET	`/health`	health_bp
GET	`/version`	health_bp
POST	`/api/chat`	chat_bp
GET	`/api/chat/models`	chat_bp
GET	`/api/session/<id>`	session_bp
DELETE	`/api/session/<id>`	session_bp
GET	`/api/sessions`	session_bp

四、Blueprint的错误处理

Blueprint 可以定义自己的错误处理器：

python

chat_bp = Blueprint("chat", __name__)


@chat_bp.errorhandler(429)
def rate_limit_exceeded(error):
    """对话接口的限流错误"""
    return {"error": "请求太频繁，请稍后再试"}, 429

Blueprint 的错误处理器只会在该蓝图的视图函数中触发。如果蓝图内部没有匹配的处理器，会继续查找应用级别的错误处理器。

常见做法是按路径前缀区分错误格式：

python

@app.errorhandler(404)
@app.errorhandler(405)
def handle_api_error(ex):
    if request.path.startswith("/api/"):
        # API接口返回JSON
        return jsonify(error=str(ex)), ex.code
    else:
        # 其他页面返回默认HTML
        return ex

五、Blueprint的URL生成

使用url_for生成 Blueprint 中的 URL 时，需要带上蓝图名前缀：

python

from flask import url_for

# 格式：蓝图名.函数名
url_for("chat.chat")           # → /api/chat
url_for("session.get_session", session_id="abc")  # → /api/session/abc
url_for("health.health")       # → /health

在 Blueprint 内部，可以使用.函数名省略当前蓝图名：

python

@chat_bp.route("/chat")
def chat():
    # 在chat_bp内部跳转到其他端点
    models_url = url_for(".list_models")  # → /api/chat/models
    return {"models_url": models_url}

六、嵌套Blueprint

Blueprint 还可以注册到另一个 Blueprint 上，用于更细粒度的模块划分：

python

# 父蓝图
api_bp = Blueprint("api", __name__, url_prefix="/api")

# 子蓝图
chat_bp = Blueprint("chat", __name__)
session_bp = Blueprint("session", __name__)

# 子蓝图注册到父蓝图
api_bp.register_blueprint(chat_bp, url_prefix="/chat")
api_bp.register_blueprint(session_bp, url_prefix="/session")

# 父蓝图注册到应用
app.register_blueprint(api_bp)

最终的URL：

路径	说明
`/api/chat/`	chat_bp的`/`路由
`/api/chat/models`	chat_bp的`/models`路由
`/api/session/<id>`	session_bp的`/<id>`路由

生成嵌套 Blueprint 的 URL 时，需要使用完整的嵌套端点名：

python

url_for("api.chat.chat")          # → /api/chat/
url_for("api.session.get_session", session_id="abc")  # → /api/session/abc

这种方式适合用于 API 版本管理：v1和v2各自作为父蓝图，下面挂载不同功能的子蓝图。

七、总结

Blueprint 的作用是将 Flask 项目从单文件结构扩展为多模块结构。

先定义后注册：在独立文件中定义路由，最后组装到应用
URL前缀：统一管理API路径，如/api/v1
模块化：每个Blueprint负责一个功能领域
错误处理：Blueprint可以有自己的错误处理器
嵌套：Blueprint可以嵌套，适合API版本管理

借助 Blueprint，Agent 项目可以按功能拆分路由模块，并在应用入口统一注册。

下一篇将介绍流式响应，用于让 Agent API 在生成完整回答之前持续返回部分内容。

03 蓝图拆分大型Agent项目 ​

一、创建Blueprint ​

1.1 Blueprint构造参数 ​

二、注册Blueprint ​

2.1 URL前缀 ​

三、拆分Agent项目 ​

3.1 chat.py — 对话接口 ​

3.2 session.py — 会话管理 ​

3.3 health.py — 健康检查 ​

3.4 errors.py — 错误处理 ​

3.5 app.py — 组装一切 ​

四、Blueprint的错误处理 ​

五、Blueprint的URL生成 ​

六、嵌套Blueprint ​

七、总结 ​

03 蓝图拆分大型Agent项目

一、创建Blueprint

1.1 Blueprint构造参数

二、注册Blueprint

2.1 URL前缀

三、拆分Agent项目

3.1 chat.py — 对话接口

3.2 session.py — 会话管理

3.3 health.py — 健康检查

3.4 errors.py — 错误处理

3.5 app.py — 组装一切

四、Blueprint的错误处理

五、Blueprint的URL生成

六、嵌套Blueprint

七、总结